package com.justech.tcp.server;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @ClassName
 * @Author: xiaofeng.yang
 * @Date: Create in 9:49 2019/12/3
 * @Description : TODO
 * @Version: 1.0
 */
public class Receiver {
    private static  final Logger logger= LoggerFactory.getLogger(Receiver.class);
    private String responseRouter;
    private Channel responseChannel;
    private ConsumerConnectionFactory consumerConnectionFactory;

    public Receiver() throws IOException, TimeoutException {
        consumerConnectionFactory = ConsumerConnectionFactory.getInstance();
    }

    public void binding(String routerkey) throws IOException {
        this.responseRouter = routerkey;
        this.responseChannel = consumerConnectionFactory.getConnection().createChannel();
        this.responseChannel.queueDeclare(this.responseRouter, true, false, false, null);
        this.responseChannel.basicQos(1);
        logger.info("bind receive channel:"+this.responseRouter);
    }

    public void close() {
        try {
            responseChannel.close();
        } catch (IOException e) {
            logger.error("close channel:",e);
        } catch (TimeoutException e) {
            logger.error("close channel:",e);
        }
    }

    public void receive(io.netty.channel.Channel channel) throws IOException {
        final io.netty.channel.Channel channel1 = channel;
        DefaultConsumer defaultConsumer = new DefaultConsumer(responseChannel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                if (channel1.isActive()){
                    ByteBuf out = Unpooled.copiedBuffer(new String(body)+"\n", CharsetUtil.UTF_8);
                    channel1.writeAndFlush(out);
                    responseChannel.basicAck(envelope.getDeliveryTag(), true);
                }else {
                    close();
                    responseChannel.basicNack(envelope.getDeliveryTag(),false,true);
                }
            }
        };
        this.responseChannel.basicConsume(this.responseRouter, false, defaultConsumer);
    }

}
